package com.ruan.consumer;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.rabbitmq.client.Channel;
import com.ruan.entity.ErrorOrder;
import com.ruan.entity.Orders;
import com.ruan.entity.Sku;
import com.ruan.entity.Spu;
import com.ruan.mapper.ErrorOrderMapper;
import com.ruan.mapper.OrdersMapper;
import com.ruan.mapper.SkuMapper;
import com.ruan.mapper.SpuMapper;
import com.ruan.utils.RedisCache;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;

/**
 * @ClassName:OrdersListener
 * @Author:阮昇
 * @Description:
 * @Datetime: 2023/9/15 14:09
 **/
@Component
public class OrdersListener {

    @Autowired
    private OrdersMapper ordersMapper;

    @Autowired
    private ErrorOrderMapper errorOrderMapper;

    @Autowired
    private SpuMapper spuMapper;

    @Autowired
    private SkuMapper skuMapper;

    @Autowired
    private RedisCache redisCache;

    @RabbitListener(queues = "pay-confirm-queue")
    public void processMessage(Message message, Channel channel) throws IOException {
        try {
            // 处理消息逻辑
            byte[] body = message.getBody();
            String bodyString = new String(body, StandardCharsets.UTF_8);
            QueryWrapper<Orders> wrapper = new QueryWrapper<>();
            wrapper.eq("order_number",bodyString);
            wrapper.eq("status",1);
            Orders orders = ordersMapper.selectOne(wrapper);
            orders.setPayFlag(1);
            int row = ordersMapper.updateById(orders);
            Sku sku = skuMapper.selectById(orders.getSkuId());
            int skuSales = sku.getSales()+orders.getNum();
            sku.setSales(skuSales);
            skuMapper.updateById(sku);
            Spu spu = spuMapper.selectById(sku.getSpuId());
            int spuSales = spu.getTotalSales()+orders.getNum();
            spu.setTotalSales(spuSales);
            spuMapper.updateById(spu);
            if (row != 1){
                ErrorOrder errorOrder = new ErrorOrder();
                errorOrder.setOrderNumber(bodyString);
                errorOrderMapper.insert(errorOrder);
            }
            // 手动确认消息处理成功
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            //处理消息失败
            // 判断消息的重试次数
            if (message.getMessageProperties().getRedelivered()) {
                // 超过重试次数，拒绝消息，并将其发送到死信队列
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                // 未超过重试次数，将消息重新投递，这里需要注意的是第二个参数true，设置为true表示这条消息在队列中所有的消费者都可以接收，是一种广播形式。
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
        }
    }


    @RabbitListener(queues = "orders-delayed-queue")
    public void consumeMessage(Message message, Channel channel) throws IOException {
        try {
            // 处理消息逻辑
            byte[] body = message.getBody();
            String bodyString = new String(body, StandardCharsets.UTF_8);
            QueryWrapper<Orders> wrapper = new QueryWrapper<>();
            wrapper.eq("order_number",bodyString);
            Orders orders = ordersMapper.selectOne(wrapper);
            if (orders.getPayFlag()==0){
                orders.setStatus(0);
                ordersMapper.updateById(orders);
                redisCache.increment("skuStock"+orders.getSkuId(), orders.getNum());
            }

            // 手动确认消息处理成功
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            //处理消息失败
            // 判断消息的重试次数
            if (message.getMessageProperties().getRedelivered()) {
                // 超过重试次数，拒绝消息，并将其发送到死信队列
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                // 未超过重试次数，将消息重新投递，这里需要注意的是第二个参数true，设置为true表示这条消息在队列中所有的消费者都可以接收，是一种广播形式。
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
        }
    }


    @RabbitListener(queues = "dead-letter-pay-confirm-queue")
    public void ordersDealMessage(Message message, Channel channel) throws IOException {
        ErrorOrder errorOrder = new ErrorOrder();
        byte[] body = message.getBody();
        String bodyString = new String(body, StandardCharsets.UTF_8);
        errorOrder.setOrderNumber(bodyString);
        System.err.println(bodyString);
        errorOrderMapper.insert(errorOrder);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }


}
